package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo22Acc {
  // 累加器
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo22Acc")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val intRDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), 3)

    var localVal: Int = 1
    println(s"map任务之前的localVal的值为：$localVal") // 1

    /**
     * 在算子内部对算子外部的值是不能进行修改的
     * 算子内部的代码需要封装成Task最终发送到Executor中执行
     * 算子外部的代码是在Driver端执行的
     * localVal变量是在Driver端定义的
     * 在Task中使用的localVal变量实际上是Driver端定义的localVal变量的一个副本
     * 所以在Task中对Driver定义的变量进行修改是无效的
     * 如果需要进行修改 可以使用累加器
     *
     */
    intRDD.map(i => {
      localVal = localVal + i
      println(s"i的值为：$i")
      println(s"localVal的值为：$localVal")
      i
    }).collect()
    println(s"map任务之后的localVal的值为：$localVal") // 1

    // 如果使用累加器

    // 在算子外部定义
    val longAcc: LongAccumulator = sc.longAccumulator("longAcc")
    println(s"map任务之前的longAcc的值为：${longAcc.value}")

    // 在算子内部使用
    intRDD.map(i => {
      // 每个分区的数据会由每个Task去处理
      // 每个Task在处理时 都会从 累加器的默认值开始
      longAcc.add(i)
      println(s"i的值为：$i")
      println(s"longAcc的值为：${longAcc.value}")
      i
    }).collect()

    // 最终在Driver端会对每个Task中计算的得到的累加器的值进行汇总 得到最后的结果
    println(s"map任务之后的longAcc的值为：${longAcc.value}")


  }

}
